package sql;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;
import java.util.UUID;

/**
 案例：窗口TopN
    按1min为窗口，统计这1min内排名前N的热词
 */
public class D12_2KafakSource_window_TopN {


    public static void main(String[] args) throws InterruptedException {


        Configuration flinkConf = new Configuration();
        flinkConf.setString("rest.port","9093");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.enableCheckpointing(10 * 1000);
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        checkpointConfig.setCheckpointStorage("file:///flink_ckp");

        String genSql = "CREATE TABLE ods_tb ( " +
            " stime STRING," +
            " word STRING," +
            " rowtime AS to_timestamp(stime)," +
            " WATERMARK FOR rowtime AS rowtime - interval '10' second" +
            ") WITH ( " +
            "  'connector' = 'kafka'," +
            "  'topic' = 'test'," +
            "  'properties.bootstrap.servers' = 'kafka:9092'," +
            "  'properties.group.id' = '" + UUID.randomUUID().toString() +"'," +
            "  'scan.startup.mode' = 'latest-offset'," +
            "  'format' = 'json'" +
            ")";


        String sinkPrint = "CREATE TABLE print (" +
            "    stime timestamp," +
            "    key STRING, " +
            "    cnt BIGINT," +
            "    rownum BIGINT" +
            ") WITH (" +
            "     'connector' = 'print'" +
            ")";

        /**
         * row_number() -> WindowRank
         */
        String sql = "INSERT INTO print " +
            " SELECT window_start, word, cnt, rownum" +
            " FROM" +
            "  (" +
            "    SELECT" +
            "      *,row_number() OVER (partition by window_start,window_end order by cnt desc ) as rownum" +
            "    FROM" +
            "      (" +
            "        SELECT window_start,window_end, word, count(1) as cnt" +
            "        FROM table (tumble (table ods_tb,descriptor (rowtime), interval '60' second))" +
            "        GROUP BY window_start, window_end, word" +
            "      )" +
            "  ) where rownum < 4";

        System.out.println(sql);

        tableEnv.executeSql(genSql);
        tableEnv.executeSql(sinkPrint);
        tableEnv.executeSql(sql);





        System.out.println(new Timestamp(System.currentTimeMillis()));



    }
}
